package study.cdc.debezium.support.debezium;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class ChangeEventHandler implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {

    @Override
    public void handleBatch(List<ChangeEvent<String, String>> records, DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
        log.info(">>>>>> {}", records.size());
//
        records.stream().forEach(e -> {
            log.info("key= {}", e.key());
            log.info("{}", e.value());
            try {
                committer.markProcessed(e);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        });
        committer.markBatchFinished();
    }
}
